package com.teemor.im.remoting.window;

import com.alibaba.fastjson.annotation.JSONField;
import com.teemor.im.common.model.ImMessage;
import com.teemor.im.discovery.node.LocalCacheNodeTable;
import com.teemor.im.discovery.node.ServerNode;
import com.teemor.im.remoting.session.LocalSession;
import com.teemor.im.remoting.session.ServerSession;
import lombok.Data;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * 会话窗口，维持一个聊天会话的用户和消息等
 *
 * @ename zhoulk
 * @cname 周李科
 * @date 2020/6/15 10:28
 */
@Data
public class SessionWindow {

    /**
     * 窗口ID
     */
    private Long id;

    /**
     * 加入窗口的用户session列表
     */
    private Map<String, ServerSession> sessionMap;

    /**
     * 消息列表
     */
    private List<ImMessage> messages;

    /**
     * 订阅了该窗口的节点列表
     */
    private Map<ServerNode, Integer> nodeMap;

    /**
     * 窗口模式，
     * 0-临时窗口：所有人退出窗口后，窗口即销毁
     * 1-持久化窗口：持久化存在，创建人不解散则会一直存在
     */
    private Integer model;

    @JSONField(serialize = false)
    private Lock lock = new ReentrantLock();


    public SessionWindow(Long windowId, Integer model) {
        this.id = windowId;
        this.sessionMap = new ConcurrentHashMap<>();
        this.messages = new ArrayList<>();
        this.nodeMap = new ConcurrentHashMap<>();
        this.model = model;
    }

    /**
     * 订阅该窗口
     *
     * @param session
     */
    public void subscribe(ServerSession session) {
        lock.lock();
        try {
            sessionMap.put(session.getId(), session);
            ServerNode serverNode = session.getServerNode();
            if (nodeMap.containsKey(serverNode)) {
                nodeMap.put(serverNode, nodeMap.get(serverNode) + 1);
            } else {
                nodeMap.put(serverNode, 1);
            }
        } finally {
            lock.unlock();
        }

    }

    public List<ServerSession> getSessions() {
        return new ArrayList<>(sessionMap.values());
    }

    public void removeSession(String sessionId) {
        lock.lock();
        try {
            if (sessionMap.containsKey(sessionId)) {
                ServerSession session = sessionMap.get(sessionId);
                sessionMap.remove(sessionId);
                ServerNode serverNode = session.getServerNode();
                Integer count = nodeMap.get(serverNode);
                if (count == null) return;
                if (--count == 0) nodeMap.remove(serverNode);
                nodeMap.put(serverNode, count);
            }
        } finally {
            lock.unlock();

        }
    }

    /**
     * 推送消息给聊天窗口所有用户，不推送给发送者
     *
     * @param serverSession
     * @param message
     */
    public void pushMessage(ServerSession serverSession, ImMessage message) {
        sessionMap.values().forEach(session -> {
            if (!serverSession.getSessionId().equals(session.getSessionId()) && serverSession instanceof LocalSession) {
                session.writeAndFlush(message);
            }
        });
        //获取所有非当前节点
        Set<ServerNode> serverNodes = LocalCacheNodeTable.getRemoteNodes().stream()
                .filter(node -> !node.equals(LocalCacheNodeTable.getLocalNode()))
                .collect(Collectors.toSet());
        serverNodes.forEach(node -> {
            node.pushMessage(message);
        });
    }

    /**
     * 其它节点消息转发
     * @param message
     */
    public void forwardMessage(ImMessage message) {
        sessionMap.values().forEach(session -> {
            session.writeAndFlush(message);
        });
    }

    /**
     * 拉取聊天窗口中所有消息
     *
     * @return
     */
    public List<ImMessage> pullMessage(ServerSession session) {
        session.writeAndFlush(messages);
        return messages;
    }

}
